package members

import (
	"context"
	"dash/collectors"
	"dash/messages"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/go-kit/kit/log"
	"github.com/go-redis/redis/v8"
	"github.com/olivere/elastic/v7"
)

// 采集器与采集用户关联的仓储服务，目前仅完善了通过采集号获取用户的操作
// 采集器与采集人目前为多对多的关系
type CollectorUserRepository interface {
	// 通过采集器id获取采集用户
	GetUserByCollectorId(ctx context.Context, id string) ([]CollectorUser, error)
}

type collectorUserRepository struct {
	SqlConn *sql.DB
	logger  log.Logger
}

func NewCollectorUserRepository(sqlConn *sql.DB, logger log.Logger) CollectorUserRepository {
	return &collectorUserRepository{SqlConn: sqlConn, logger: logger}
}

// 通过采集号Id获取所有关联的采集人对象
func (r *collectorUserRepository) GetUserByCollectorId(ctx context.Context, id string) ([]CollectorUser, error) {
	var users []CollectorUser
	rows, err := r.SqlConn.Query(`
		SELECT 
			user.id,
			user.name,
			organization.name
		FROM collector_user
		INNER JOIN user
		INNER JOIN organization
		ON collector_user.userId = user.id and user.organId = organization.id
		WHERE collector_user.collectorId = ?
	`,
		id)
	// Query方法调用之后需要调用Close接口
	defer func() {
		err := rows.Close()
		if err != nil {
			_ = r.logger.Log("mysql error", err.Error())
		}
	}()
	if err != nil {
		_ = r.logger.Log
		return []CollectorUser{}, nil
	}
	for rows.Next() {
		var u CollectorUser
		var organ organization
		err := rows.Scan(
			&u.Id,
			&u.Name,
			&organ.Name)
		if err != nil {
			_ = r.logger.Log("mysql error", err.Error())
			continue
		}
		u.Organization = organ
		users = append(users, u)
	}
	return users, nil
}

// 网民相关的仓储操作
type Repository interface {
	// 通过id获取member
	Get(ctx context.Context, id string) (messages.Member, error)
	// 通过key关联pipeline中的计数器获取key相关的7天发言量
	GetMessageCount7(ctx context.Context, key string) (int64, error)
	// 获取网民最新的发言事件
	GetLastMessageTimestamp(ctx context.Context, id string) (int64, error)
	// 获取网民7天内的预警消息量
	GetAlarmMessageCount7(ctx context.Context, id string) (int64, error)
	// 保存网民相关信息(仅可修改网民可编辑部分)
	Save(ctx context.Context, member messages.Member) error
}

// member关联mysql相关的数据库对象
type MemberRow struct {
	// 网民id
	Id               string
	// 网民昵称
	NickName         sql.NullString
	// 网民年龄
	Age              sql.NullInt64
	// 网民性别，使用字符串
	Sex              sql.NullString
	// 网民头像的url地址
	Avatar           sql.NullString
	// 网民被捕获的时间戳
	CaptureTimestamp sql.NullInt64
	// 网民数据来源：微信/qq
	DataSource       sql.NullString
	// 是否关注
	Watched          sql.NullBool
	// 备注
	Comment          sql.NullString
	// 姓名
	Name             sql.NullString
	// 电子邮件
	Email            sql.NullString
	// 身份证
	IdCard           sql.NullString
	// 身份证照片url地址
	IdCardAvatar     sql.NullString
	// 常住地
	ResidenceAddress sql.NullString
	// 户籍所在地
	CensusAddress    sql.NullString
	// 个性签名
	Signature        sql.NullString
	// 更新时间戳
	UpdateTimestamp  sql.NullInt64
	// 历史个性签名， []byte是由于内部都存储为json格式
	UsedSignature    []byte
	// 历史昵称
	UsedNickname     []byte
	// 标签
	Tags             []byte
	// 历史姓名
	UsedName         []byte
	// 电话号
	PhoneNumber      []byte
	// 居住地 eg：北京 朝阳
	Address          []byte
}

type chatroomRow struct {
	Id         string
	Name       sql.NullString
	DataSource sql.NullString
	Avatar     sql.NullString
	ownerId    sql.NullString
}

type defaultRepository struct {
	SqlConn   *sql.DB
	RedisConn *redis.Client
	EsConn    *elastic.Client
	logger    log.Logger
}

// 获取网民最新发言时间戳
func (d *defaultRepository) GetLastMessageTimestamp(ctx context.Context, id string) (int64, error) {
	return d.RedisConn.Get(ctx, fmt.Sprintf("memberLatestActiveTime:%s", id)).Int64()
}

// 通过es获取网民7天预警数量
func (d *defaultRepository) GetAlarmMessageCount7(ctx context.Context, id string) (int64, error) {
	results, err := d.EsConn.Search("alarm_message").
		Query(elastic.NewBoolQuery().
			Must(elastic.NewMatchQuery("member.id", id)).
			Filter(elastic.NewRangeQuery("timestamp").
				Gte(time.Now().Unix() - 60*60*24*7))).Do(ctx)
	if err != nil {
		_ = d.logger.Log("es search member alarm count error:", err)
		return 0, err
	}
	count := results.Hits.TotalHits.Value
	return count, nil
}

// sql获取群主
func (d *defaultRepository) getOwner(_ context.Context, ownerId, chatroomId string) (messages.Member, error) {
	ownerRow := MemberRow{Id: ownerId}
	var ownerAlias sql.NullString
	err := d.SqlConn.QueryRow(`
			SELECT 
				member.Nickname,
				chatroom_member.alias
			FROM chatroom_member
			INNER JOIN member
			ON member.id = chatroom_member.memberId
			WHERE 
				chatroom_member.chatroomId = ? and 
				chatroom_member.memberId = ?;`,
		chatroomId,
		ownerId).
		Scan(
			&ownerRow.NickName,
			&ownerAlias)
	if err != nil {
		if err == sql.ErrNoRows {
			return messages.Member{}, errors.New("id 不存在")
		} else {
			_ = d.logger.Log("get member scan error:", err)
			return messages.Member{}, errors.New("服务器内部错误")
		}
	}
	owner := memberRowToMember(ownerRow)
	if ownerAlias.Valid {
		owner.Aliases = []string{ownerAlias.String}
	} else {
		owner.Aliases = []string{owner.NickName}
	}
	return owner, nil
}

// 查询出所有群
func (d *defaultRepository) getChatrooms(ctx context.Context, id, nickname string) ([]messages.Chatroom, []string, error) {
	var crs []messages.Chatroom
	var aliases []string
	crows, err := d.SqlConn.Query(`
		SELECT 
			chatroom.name, 
			chatroom.id, 
			chatroom.ownerId, 
			chatroom_member.alias
    	FROM member 
    	INNER JOIN chatroom_member 
    	INNER JOIN chatroom 
		ON 
			member.id = chatroom_member.memberId 
			and chatroom_member.quit = false 
			and chatroom_member.chatroomId = chatroom.id
    	WHERE member.id = ?;
	`, id)
	defer func() {
		err := crows.Close()
		if err != nil {
			_ = d.logger.Log("sql error", err.Error())
		}
	}()
	for crows.Next() {
		var alias sql.NullString
		var chatroomRow chatroomRow
		var ownerRow MemberRow
		var owner messages.Member
		err = crows.Scan(
			&chatroomRow.Name,
			&chatroomRow.Id,
			&ownerRow.Id,
			&alias,
		)
		if err != nil || len(chatroomRow.Id) == 0 {
			continue
		}
		owner, err = d.getOwner(ctx, ownerRow.Id, chatroomRow.Id)
		if err != nil {
			_ = d.logger.Log("mysql error owner", err.Error())
		}
		// 如果没有群昵称则为微信昵称
		if alias.Valid {
			aliases = append(aliases, alias.String)
		} else {
			aliases = append(aliases, nickname)
		}

		chatroom := ToChatroom(chatroomRow, owner)
		chatroom.Owner = owner
		crs = append(crs, chatroom)
	}
	return crs, aliases, nil
}

// sql查询出网民关联的所有采集号
func (d *defaultRepository) getCollectors(_ context.Context, id string) ([]collectors.Collector, error) {
	colRows, err := d.SqlConn.Query(`
		SELECT
			collector.id,
			collector.name
		FROM collector
		INNER JOIN collector_member ON 
			collector_member.collectorId = collector.id
		WHERE collector_member.memberId = ?;
	`, id)
	if err != nil {
		return nil, err
	}
	defer func() {
		err := colRows.Close()
		if err != nil {
			_ = d.logger.Log("Mysql error:", err.Error())
		}
	}()
	var cols []collectors.Collector
	for colRows.Next() {
		var collectorId sql.NullString
		var collectorName sql.NullString
		err := colRows.Scan(&collectorId, &collectorName)
		if err != nil {
			_ = d.logger.Log("Mysql error:", err.Error())
			continue
		}
		var col collectors.Collector
		if collectorId.Valid {
			col.Id = collectorId.String
		}
		if collectorName.Valid {
			col.Name = collectorName.String
		}
		cols = append(cols, col)
	}
	return cols, nil
}

// 获取member对象的仓储方法
// 调用getChatrooms获取网民所在的群信息
// 调用getCollector获取网民关联的所有采集号
func (d *defaultRepository) Get(ctx context.Context, id string) (messages.Member, error) {
	var memberRow MemberRow
	err := d.SqlConn.QueryRow(
		`
		SELECT
			id,
			nickname,
			age,
			sex,
			avatar,
			captureTimestamp,
			usedNickname,
			dataSource,
			watched,
			tags,
			signature,
			comment,
			usedName,
			name,
			email,
			phoneNumber,
			idCard,
			idCardAvatar,
			address,
			residenceAddress,
			censusAddress,
			usedSignature
		FROM member WHERE id = ?`, id).Scan(
		&memberRow.Id,
		&memberRow.NickName,
		&memberRow.Age,
		&memberRow.Sex,
		&memberRow.Avatar,
		&memberRow.CaptureTimestamp,
		&memberRow.UsedNickname,
		&memberRow.DataSource,
		&memberRow.Watched,
		&memberRow.Tags,
		&memberRow.Signature,
		&memberRow.Comment,
		&memberRow.UsedName,
		&memberRow.Name,
		&memberRow.Email,
		&memberRow.PhoneNumber,
		&memberRow.IdCard,
		&memberRow.IdCardAvatar,
		&memberRow.Address,
		&memberRow.ResidenceAddress,
		&memberRow.CensusAddress,
		&memberRow.UsedSignature,
	)
	if err != nil {
		if err == sql.ErrNoRows {
			return messages.Member{}, errors.New("id 不存在")
		} else {
			_ = d.logger.Log("get member scan error:", err)
			return messages.Member{}, errors.New("服务器内部错误")
		}
	}
	member := memberRowToMember(memberRow)

	var aliases []string
	chatrooms, aliases, _ := d.getChatrooms(ctx, id, member.NickName)
	member.Chatrooms = chatrooms
	member.Aliases = aliases

	cols, _ := d.getCollectors(ctx, id)
	member.Collectors = cols
	return member, nil
}

func ToChatroom(r chatroomRow, owner messages.Member) messages.Chatroom {
	var chatroom messages.Chatroom
	chatroom.Id = r.Id
	if r.Name.Valid {
		chatroom.Name = r.Name.String
	}
	if r.DataSource.Valid {
		chatroom.DataSource = r.DataSource.String
	}
	if r.Avatar.Valid {
		chatroom.Avatar = r.Avatar.String
	}
	if r.ownerId.Valid {
		chatroom.Owner = owner
	}
	return chatroom
}

// 获取key关联的七天的发言量，key可传入网民也可传入群
func (d *defaultRepository) GetMessageCount7(ctx context.Context, key string) (int64, error) {
	redisKey := "%s:%s"
	today := time.Now()
	var total int64
	for i := 0; i > -7; i-- {
		year, month, day := today.AddDate(0, 0, i).Date()
		date := fmt.Sprintf("%d%02d%02d", year, int(month), day)
		totalKey := fmt.Sprintf(redisKey, key, date)
		dailyCount, err := d.RedisConn.Get(ctx, totalKey).Int64()
		if err != nil {
			return total, err
		}
		total += dailyCount
	}
	return total, nil
}

// 保存群对象的sql实现
func (d *defaultRepository) Save(_ context.Context, member messages.Member) error {
	memberRow := memberToMemberRow(member)
	memberRow.UpdateTimestamp = sql.NullInt64{Int64: time.Now().UnixNano() / 1e6, Valid: true}
	_, err := d.SqlConn.Exec(`
		UPDATE member
		SET 
			tags = ?,
			name = ?,
			age = ?,
			sex = ?,
			phoneNumber = ?,
			idCard = ?,
			censusAddress = ?,
			residenceAddress = ?,
			comment = ?,
			updateTimestamp = ?
		WHERE id = ?
		`,
			memberRow.Tags,
			memberRow.Name,
			memberRow.Age,
			memberRow.Sex,
			memberRow.PhoneNumber,
			memberRow.IdCard,
			memberRow.CensusAddress,
			memberRow.ResidenceAddress,
			memberRow.Comment,
			memberRow.UpdateTimestamp,
			memberRow.Id)
	if err != nil{
		return err
	}
	return nil
}

func NewMemberRepository(conn *sql.DB, logger log.Logger, redisConn *redis.Client, esConn *elastic.Client) Repository {
	return &defaultRepository{
		SqlConn:   conn,
		RedisConn: redisConn,
		EsConn:    esConn,
		logger:    logger,
	}
}

func memberToMemberRow(member messages.Member) MemberRow{
	var memberRow MemberRow
	tags, _ := json.Marshal(member.Tags)
	if tags != nil{
		memberRow.Tags = tags
	}
	memberRow.Id = member.Id
	memberRow.Name = sql.NullString{String: member.Name, Valid: true}
	memberRow.Age = sql.NullInt64{Int64: member.Age, Valid: true}
	memberRow.Sex = sql.NullString{String: member.Sex, Valid: true}
	memberRow.IdCard = sql.NullString{String: member.IdCard, Valid: true}
	phoneNumber, _ := json.Marshal(member.PhoneNumber)
	if phoneNumber != nil{
		memberRow.PhoneNumber = phoneNumber
	}
	memberRow.CensusAddress = sql.NullString{String: member.CensusAddress, Valid: true}
	memberRow.ResidenceAddress = sql.NullString{String: member.ResidenceAddress, Valid: true}
	memberRow.Comment = sql.NullString{String: member.Comment, Valid: true}
	return memberRow
}

func memberRowToMember(row MemberRow) messages.Member {
	member := messages.Member{}
	member.Id = row.Id
	if row.NickName.Valid {
		member.NickName = row.NickName.String
	}
	if row.Age.Valid {
		member.Age = row.Age.Int64
	}
	if row.Sex.Valid {
		member.Sex = row.Sex.String
	}
	if row.Avatar.Valid {
		member.Avatar = row.Avatar.String
	}
	if row.CaptureTimestamp.Valid {
		member.CaptureTimestamp = row.CaptureTimestamp.Int64
	}
	if row.DataSource.Valid {
		member.DataSource = row.DataSource.String
	}
	if row.Watched.Valid {
		member.Watched = row.Watched.Bool
	}
	if row.Signature.Valid {
		member.Signature = row.Signature.String
	}
	if row.Comment.Valid {
		member.Comment = row.Comment.String
	}
	if row.Name.Valid {
		member.Name = row.Name.String
		member.UsedName = append(member.UsedName, row.Name.String) //member曾用名第一个元素为真实姓名
	}
	if row.Email.Valid {
		member.Email = row.Email.String
	}
	if row.IdCard.Valid {
		member.IdCard = row.IdCard.String
	}
	if row.IdCardAvatar.Valid {
		member.IdCardAvatar = row.IdCardAvatar.String
	}
	if row.ResidenceAddress.Valid {
		member.ResidenceAddress = row.ResidenceAddress.String
	}
	if row.CensusAddress.Valid {
		member.CensusAddress = row.CensusAddress.String
	}

	var tags []string
	err := json.Unmarshal(row.Tags, &tags)
	if err != nil {
		_ = err
	}
	member.Tags = tags

	var usedNames []string
	err = json.Unmarshal(row.UsedName, &usedNames)
	if err != nil {
		_ = err
	}
	member.UsedName = usedNames

	var phoneNumbers []string
	err = json.Unmarshal(row.PhoneNumber, &phoneNumbers)
	if err != nil {
		_ = err
	}

	member.PhoneNumber = phoneNumbers

	var address []string
	err = json.Unmarshal(row.Address, &address)
	if err != nil {
		_ = err
	}
	member.Address = address

	usedNickname := make([]string, 0, 3)
	err = json.Unmarshal(row.UsedNickname, &usedNickname)
	if err != nil {
		_ = err
	}
	member.UsedNickname = usedNickname

	usedSignatures := make([]string, 0, 3)
	err = json.Unmarshal(row.UsedSignature, &usedSignatures)
	if err != nil {
		_ = err
	}
	member.UsedSignature = usedSignatures
	return member
}

// BUG(topics) memberRow.UpdateTimestamp时间戳应该以ms为单位,在save方法中应该计算并除1e6